/**
 * @AccessMessageDecoder.java  COPYRIGHT xiaojian
 * 
 * Created on 2017年7月5日 下午2:57:47
 */
package xj.toolkit.access.acceptor.codec;

import java.util.List;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.ByteToMessageDecoder;
import io.netty.util.AttributeKey;
import xj.toolkit.bean.CommonsHeader;
import xj.toolkit.util.ByteUtils;
import xj.toolkit.access.AccessHeader;
import xj.toolkit.access.AccessMsgCodeConstants;
import xj.toolkit.access.msg.AccessAckMessage;
import xj.toolkit.access.msg.AccessContentMessage;

/**
 * 将byte的数组转为对象供系统处理。
 * 
 * @author <a href="mailto:cxj2000@gmail.com">xiaojian.cao</a>
 *
 */
public class AccessMessageDecoder extends ByteToMessageDecoder {

	private static final Logger logger = LoggerFactory.getLogger(AccessMessageDecoder.class);

	private static final AttributeKey<CommonsHeader> HEADER_ATTRIBUTE_KEY = AttributeKey
			.valueOf("XIP_HEADER_ATTRIBUTE_KEY");

	// 大于1M的数据包可能是问题数据包
	private int maxMessageLength = 1024 * 1024;

	private int dumpBytes = 256;

	private boolean isDebugEnabled;

	public int getMaxMessageLength() {
		return maxMessageLength;
	}

	public void setMaxMessageLength(int maxMessageLength) {
		this.maxMessageLength = maxMessageLength;
	}

	public int getDumpBytes() {
		return dumpBytes;
	}

	public void setDumpBytes(int dumpBytes) {
		this.dumpBytes = dumpBytes;
	}

	public boolean isDebugEnabled() {
		return isDebugEnabled;
	}

	public void setDebugEnabled(boolean isDebugEnabled) {
		this.isDebugEnabled = isDebugEnabled;
	}

	@Override
	protected void decode(ChannelHandlerContext ctx, ByteBuf buffer, List<Object> out) throws Exception {
		CommonsHeader header = ctx.channel().attr(HEADER_ATTRIBUTE_KEY).get();
		if (null == header) {
			if (buffer.readableBytes() < CommonsHeader.HEADER_LENGTH) {
				return;
			}

			header = new CommonsHeader();
			header.setBasicVer(buffer.readByte());
			header.setPayloadType(buffer.readShort());
			header.setType(buffer.readShort());
			header.setFirstTransaction(buffer.readLong());
			header.setSecondTransaction(buffer.readLong());
			header.setReserved(buffer.readInt());
			header.setMessageCode(buffer.readInt());
			header.setMessageLength(buffer.readInt());

			if (logger.isDebugEnabled() && isDebugEnabled) {
				logger.debug("header-->" + header);
			}

			if (header.getMessageLength() < 0) {
				logger.error("header.messageLength ({}) < 0, so drop this connection {}.", header.getMessageLength(),
						ctx.channel());
				ctx.channel().close().await();
				throw new RuntimeException("messageLength too large:" + header.getMessageLength());
			}
			if (maxMessageLength > 0 && header.getMessageLength() > maxMessageLength) {
				logger.error("header.messageLength ({}) exceed maxMessageLength[{}], so drop this connection{}.",
						header.getMessageLength(), maxMessageLength, ctx.channel());
				ctx.channel().close().await();
				throw new RuntimeException("messageLength too large:" + header.getMessageLength());
			}

			ctx.attr(HEADER_ATTRIBUTE_KEY).set(header);
		}

		if (buffer.readableBytes() < header.getMessageLength()) {
			logger.debug("readableBytes {} smaller than packageLength {},waiting for remain bytes",
					buffer.readableBytes(), header.getMessageLength());
			return;
		}

		// 为下一次在同一ctx上进行xip接受初始化环境
		ctx.attr(HEADER_ATTRIBUTE_KEY).remove();

		AccessHeader accessHeader = new AccessHeader();
		accessHeader.setSourceModuleId(buffer.readShort());
		accessHeader.setTargetInstanceId(buffer.readShort());
		accessHeader.setTargetModuleId(buffer.readShort());
		accessHeader.setTargetInstanceId(buffer.readShort());
		accessHeader.setReserved(buffer.readInt());

		// -- 如果是ACK信息
		if (header.getMessageCode() == AccessMsgCodeConstants.MSG_CODE_ACK) {
			AccessAckMessage ackMessage = new AccessAckMessage();
			ackMessage.setAccessHeader(accessHeader);
			ackMessage.setCommonHeader(header);

			if (logger.isDebugEnabled()) {
				logger.debug("message[{}].", ackMessage);
			}

			out.add(ackMessage);
			return;
		}

		final byte[] bytes = new byte[header.getMessageLength() - AccessHeader.HEADER_LENGTH];
		buffer.readBytes(bytes);
		if (logger.isTraceEnabled() && isDebugEnabled) {
			logger.trace("body raw bytes \r\n{}", ByteUtils.bytesAsHexString(bytes, dumpBytes));
		}

		AccessContentMessage message = new AccessContentMessage();
		message.setCommonHeader(header);
		message.setAccessHeader(accessHeader);
		message.setContent(bytes);

		if (logger.isDebugEnabled()) {
			logger.debug("message[{}].", message);
		}

		out.add(message);

	}

}
